package com.iot.ruler.caster;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 数据分发
 * 处理发布者的数据发布
 * 处理订阅者的订阅
 * 处理主题(Subject)与订阅者的绑定
 *
 * @author lzb
 * @Date 2023/2/17 14:35
 */
public class RulerCaster {
    private static final RulerCaster CASTER = new RulerCaster();
    /**
     * 无锁集合，用于存放观察者，由于无锁操作
     * 本库内不会进行同时读写
     */
    private final static List<RulerSubscriber<?>> OBSERVERS = new ArrayList<>();
    /**
     * 单线程线程池
     * 线程异常退出时能够新建线程替代
     * 使用阻塞队列进行缓存
     * 此处的线程队列设置为1，数据已经在另一个队列中进行单独缓存
     */
    private final ExecutorService EXECUTOR = new ThreadPoolExecutor(
            1, 1,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(1), r -> {
        Thread thread = new Thread(r);
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.setDaemon(true);
        thread.setName("Ruler-Caster");
        return thread;
    });
    /**
     * 主题缓冲队列，最大Integer.MAXVALUE
     */
    private final LinkedBlockingQueue<RulerSubject<?>> SUBJECTS = new LinkedBlockingQueue<>();

    public RulerCaster() {
        // 启动数据分发独立线程
        EXECUTOR.execute(() -> {
            while (!EXECUTOR.isTerminated()) {
                try {
                    RulerSubject<?> subject = SUBJECTS.take();
                    for (RulerSubscriber<?> observer : OBSERVERS) {
                        if (observer.getPublisherName().equals(subject.getClassName())) {
                            observer.accept(subject);
                        }
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }

    public static RulerCaster getInstance() {
        return CASTER;
    }

    /**
     * 注册监听器
     *
     * @param rulerSubscriber 订阅者
     */
    static void register(RulerSubscriber<?> rulerSubscriber) {
        if (!OBSERVERS.contains(rulerSubscriber)) {
            OBSERVERS.add(rulerSubscriber);
        }
    }

    /**
     * 移除注册监听器
     *
     * @param rulerSubscriber 订阅者
     */
    static void unregister(RulerSubscriber<?> rulerSubscriber) {
        OBSERVERS.remove(rulerSubscriber);
    }

    public int observersSize() {
        return OBSERVERS.size();
    }

    /**
     * 发布事件，只有SimpleSubject的监听者才能收到数据
     *
     * @param source 发布的数据
     */
    void publish(Object source) {
        publish(new SimpleRulerSubject(source));
    }

    /**
     * 发布事件
     *
     * @param subject 事件类型
     * @param <T>     事件数据类型
     */
    <T> void publish(RulerSubject<T> subject) {
        SUBJECTS.add(subject);
    }

}
